package com.pk.flink.source;

import com.pk.flink.bean.Account;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;

import java.sql.*;
import java.text.SimpleDateFormat;

public class MySqlParallelStudentSource implements ParallelSourceFunction<Account> {
    private boolean isRunning = true;


    @Override
    public void run(SourceContext<Account> ctx) throws Exception {
        Connection conn = null;
        Statement stat = null;
        ResultSet rs = null;

        try {
            Class.forName("com.mysql.jdbc.Driver");
            String connString = "jdbc:mysql://rr-2ze31wub1o9p93310ao.mysql.rds.aliyuncs.com:3306/yunmai_clouddesktop_prod?user=kjvs_dw_prod&password=TnFl6FZ2mD4PlP0O&useUnicode=true&characterEncoding=utf8&allowMultiQueries=true";
            conn = (Connection) DriverManager.getConnection(connString);

            stat = (Statement) conn.createStatement();

            String sql = "select * from uc_account limit 0,50";

            rs = (ResultSet) stat.executeQuery(sql);

            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

            while (rs.next()) {
                String guid = rs.getString("guid");
                String name = rs.getString("name");
                String createdTime = sdf.format(rs.getTimestamp("created_time"));

                Account account = new Account(guid,name,createdTime);
                ctx.collect(account);
            }

            rs.close();
            rs = null;


        } catch (SQLException e1) {
            e1.printStackTrace();

        } catch (Exception e1) {
            e1.printStackTrace();

        } finally {
            if (rs != null)
                try {
                    rs.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            if (stat != null)
                try {
                    stat.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            if (conn != null)
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
